package com.hxgis.flow.flow;

import com.hxgis.common.entity.Station;
import com.hxgis.common.entity.StationExtremeInfo;
import com.hxgis.common.enums.PreYear30Enum;
import com.hxgis.common.service.*;
import com.hxgis.common.utils.RectField;
import com.hxgis.common.utils.TimeUtil;
import com.hxgis.flow.handler.YearStationDivideHandler;
import com.iias.common.context.Tuple;
import com.iias.task.make.MakeFlow;
import com.iias.task.make.MakeNode;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.text.ParseException;
import java.util.LinkedHashMap;
import java.util.List;

@Slf4j
@Component
public class InitFlow extends MakeFlow {

    @Autowired
    RainDayService rainDayService;

    @Autowired
    private RainstormInfoService rainstormInfoService;

    @Autowired
    private StationExtremeService stationExtremeService;

    @Autowired
    RainstormCourseService rainstormCourseService;

    @Autowired
    RainstormFeatureService rainstormFeatureService;

    @Autowired
    private RainExtremeInfoService extremeInfoService;

    @Autowired
    ProcRainInfoService procRainInfoService;

    @Autowired
    ExtremeProcService extremeProcService;


    @Autowired
    RainstormCountService rainstomCountService;

    @Autowired
    StationService stationService;

    @Resource
    YearStationDivideHandler yearStationDivideHandler;

    public InitFlow() {
        super("init", "初始化所有数据");
    }

    @Override
    public void creatFlow() {

        MakeNode beginNode = begin("rainfall-data", "时间与站点纬度切分任务", yearStationDivideHandler);

        MakeNode getCimissNode = beginNode.serial("cimiss-get", "cimiss拉取元数据", "#{index} == 1 ", (tuple) -> {
            String start = tuple.getFirst();
            String end = tuple.getSecond();
            List<String> sNosList = tuple.getThird();
            // todo cimiss 拉去
            log.info("{}到{}-----cimiss拉取元数据", start, end);

            rainDayService.addToRainEs(TimeUtil.preDay(start, 1), TimeUtil.nextDay(end, 1), sNosList);
            return Tuple.addTuple(tuple);
        });

        getCimissNode.parallel("single-08", "单日或多日08", "#{index} == 1 ", (tuple) -> {
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            List<String> stations = tuple.getThird();
            // 处理单日或多日连续降水
            log.info("{}到{}-----单日或多日08", startTime, endTime);
            //单日多日降水
            rainDayService.completionRainDay08(startTime, endTime, stations);
            return Tuple.addTuple(tuple);
        }).addSyncId("cimiss-get");

        getCimissNode.parallel("single-20", "单日或多日20", "#{index} == 1 ", (tuple) -> {
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            List<String> stations = tuple.getThird();
            // 处理单日或多日连续降水
            log.info("{}到{}-----单日或多日20", startTime, endTime);
            //单日多日降水
            rainDayService.completionRainDay20(startTime, endTime, stations);
            return Tuple.addTuple(tuple);
        }).addSyncId("cimiss-get");

        getCimissNode.parallel("continue-08", "连续降水08", "#{index} == 1 ", (tuple) -> {
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            List<String> stations = tuple.getThird();
            // 处理单日或多日连续降水
            log.info("{}到{}-----连续降水08", startTime, endTime);
            //连续降水和连续无降水过程
            rainDayService.completionContinueByYear08(startTime, endTime, stations);
            return Tuple.addTuple(tuple);
        }).addSyncId("cimiss-get");

        getCimissNode.parallel("continue-20", "连续降水20", "#{index} == 1 ", (tuple) -> {
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            List<String> stations = tuple.getThird();
            // 处理单日或多日连续降水
            log.info("{}到{}-----连续降水20", startTime, endTime);
            //连续降水和连续无降水过程
            rainDayService.completionContinueByYear20(startTime, endTime, stations);
            return Tuple.addTuple(tuple);
        }).addSyncId("cimiss-get");

        getCimissNode.parallel("not-continue-20", "连续无降水20", "#{index} == 1 ", (tuple) -> {
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            List<String> stations = tuple.getThird();
            // 处理单日或多日连续降水
            log.info("{}到{}-----连续无降水20", startTime, endTime);
            //连续降水和连续无降水过程
            rainDayService.completionNotContinueByYear20(startTime, endTime, stations);
            return Tuple.addTuple(tuple);
        }).addSyncId("cimiss-get");

        getCimissNode.parallel("not-continue-08", "连续无降水08", "#{index} == 1 ", (tuple) -> {
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            List<String> stations = tuple.getThird();
            // 处理单日或多日连续降水
            log.info("{}到{}-----连续无降水08", startTime, endTime);
            //连续降水和连续无降水过程
            rainDayService.completionNotContinueByYear08(startTime, endTime, stations);
            return Tuple.addTuple(tuple);
        }).addSyncId("cimiss-get");

        MakeNode initRainStormNode = beginNode.serial("initRainStorm", "将降水数据导入暴雨表","#{index} == 1 ", (tuple -> {
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            List<String> stationNos = tuple.getThird();
            //保存暴雨数据
            rainstormInfoService.saveStorms(startTime, endTime, stationNos);
            return Tuple.addTuple(tuple);
        })).addSyncId("cimiss-get");

        // todo 这里有点问题 需要上面 单日多日 那边六个过程计算完再走这里
        MakeNode stationExtremeNode = beginNode.serial("stationExtreme", "初始化单站阈值","#{index} == 3 ", (tuple -> {
            List<String> stationIds = tuple.getFifth();
            stationExtremeService.saveExtremeInfo(PreYear30Enum.PRE_YEAR_30_START.getYear() + "0101",
                    PreYear30Enum.PRE_YEAR_30_END.getYear() + "1231", stationIds);
            return Tuple.addTuple(tuple);
        })).addSyncId("single-08","single-20","continue-08","continue-20","not-continue-20","not-continue-08");

        // 计算单站过程
        MakeNode calcSingleStormNode = initRainStormNode.parallel("calcSingleStorm", "计算单站过程","#{index} == 2 ", (tuple -> {
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            //算单站过程
            rainstormInfoService.transSingleStrom(startTime, endTime);
            return Tuple.addTuple(tuple);
        })).addSyncId("initRainStorm");

        // 计算暴雨组单站过程
        MakeNode countRainStormGroupNode = initRainStormNode.parallel("countRainStormGroup", "计算暴雨组单站过程", "#{index} == 2 ", (tuple -> {
            List<Station> stationList = stationService.selectAll();
            List<String> stationNoList = RectField.getBeanFields(stationList, "stationNo");
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            // 计算暴雨组单站过程
            rainstormCourseService.countRainStormGroup(startTime, endTime, stationNoList);
            return Tuple.addTuple(tuple);
        })).addSyncId("initRainStorm");


        //计算持续性暴雨区域过程
        MakeNode countRainStormAreaNode = calcSingleStormNode.serial("countRainStormArea", "持续性暴雨区域过程计算", "#{index} == 2 ", (tuple -> {
            List<Station> stationList = stationService.selectAll();
            List<String> stationNoList = RectField.getBeanFields(stationList, "stationNo");
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            // 计算持续性暴雨过程
            rainstomCountService.countMain(startTime, endTime, stationNoList);
            return Tuple.addTuple(tuple);
        })).addSyncId("calcSingleStorm");


        //计算全国暴雨组区域过程
        MakeNode countRainStormAreaGroupAreaNode = countRainStormGroupNode.serial("countRainStormAreaGroupArea", "计算全国暴雨组区域过程",  "#{index} == 2 ", (tuple -> {
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            // 计算持续性暴雨过程
            rainstormCourseService.countRainStormAreaGroup(startTime, endTime);
            return Tuple.addTuple(tuple);
        })).addSyncId("countRainStormGroup");


        //全国持续性暴雨特征分析单站过程
        MakeNode transSingleStromFeatureNode = initRainStormNode.parallel("transSingleStrom-Feature", "全国持续性暴雨特征分析单站过程", "#{index} == 2 ", (tuple -> {
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            //算单站过程
            rainstormFeatureService.transSingleStromByDay(startTime, endTime);
            return Tuple.addTuple(tuple);
        })).addSyncId("initRainStorm");


        //全国持续性暴雨特征分析区域过程
        MakeNode featureAreaStromNode = initRainStormNode.parallel("featureAreaStromNode", "全国持续性暴雨特征分析区域过程", "#{index} == 2 ", (tuple -> {
            List<Station> stationList = stationService.selectAll();
            List<String> stationNoList = RectField.getBeanFields(stationList, "stationNo");
            String startTime = tuple.getFirst();
            String endTime = tuple.getSecond();
            // 全国持续性暴雨特征分析区域过程
            rainstormFeatureService.featureAreaStrom(startTime, endTime, stationNoList);
            return Tuple.addTuple(tuple);
        })).addSyncId("initRainStorm");




        //todo 跑极端事件数据
        MakeNode saveExtremeInfoNode = beginNode.serial("saveExtremeInfo", "跑极端事件数据", "#{index} == 2 ", (tuple) -> {
            List<StationExtremeInfo> allExtreInfo = stationExtremeService.getAllExtreInfo();
            String start = tuple.getFirst();
            String end = tuple.getSecond();
            extremeInfoService.saveExtremeInfo(start, end, allExtreInfo);
            return Tuple.addTuple(tuple);
        }).addSyncId("stationExtreme");

        //筛选出降水表原数据的过程数据出来
        MakeNode saveProRainInfoByTimeNode = saveExtremeInfoNode.serial("saveProRainInfoByTime", "筛选出降水表原数据的过程数据出来", "#{index} == 2 ", (tuple) -> {
            String start = tuple.getFirst();
            String end = tuple.getSecond();
            procRainInfoService.saveProRainInfoByTime(start, end);
            return Tuple.addTuple(tuple);
        }).addSyncId("saveExtremeInfo");

        // 极端过程数据
        MakeNode saveExtremeProcByTimeNode = saveProRainInfoByTimeNode.serial("saveExtremeProcByTime", "初始化过程数据","#{index} == 2 ", (tuple) -> {
            String start = tuple.getFirst();
            String end = tuple.getSecond();
            List<StationExtremeInfo> allExtreInfo = stationExtremeService.getAllExtreInfo();
            extremeProcService.saveExtremeProcByTime(start, end, allExtreInfo);
            return Tuple.addTuple(tuple);
        }).addSyncId("saveProRainInfoByTime");



    }
}
